﻿#pragma once

#include "atomic.h"
#include <array>
#include <sys/mman.h>
#include <unistd.h>
//#include "/home/andinghu/demo/test/test/ull_utils/ull_atomic.h"


#define CACHELINE_SIZE_BYTES 64
///#define CACHELINE_PADDING_FOR_ATOMIC_INT64_SIZE (CACHELINE_SIZE_BYTES - sizeof(std::atomic_int64_t))
#define CACHELINE_PADDING_FOR_ATOMIC_INT64_SIZE (CACHELINE_SIZE_BYTES - sizeof(atomic_int64))
#define CACHELINE_PADDING_FOR_INT64_SIZE (CACHELINE_SIZE_BYTES - sizeof(int64_t))


#define PAYLOAD_BYTES (72 - CACHELINE_SIZE_BYTES)
#define CACHELINE_PADDING_FOR_SHORT_SIZE (CACHELINE_SIZE_BYTES - sizeof(short) - PAYLOAD_BYTES)

struct DisruptorEntry
{
	short   m_payload_len;
	char    m_payload[PAYLOAD_BYTES];
	char    m_padding[CACHELINE_PADDING_FOR_SHORT_SIZE];

	bool empty()
	{
		return (0 >= m_payload_len);
	}
};




//对std::atomic_int64_t进行了封装，内存补齐保证_seq在一个缓存行中
class AtomicSequence
{
public:
	AtomicSequence(int64_t num = 0L) : _seq(num) {};
	~AtomicSequence() {};
	AtomicSequence(const AtomicSequence&) = delete;
	AtomicSequence(const AtomicSequence&&) = delete;
	void operator=(const AtomicSequence&) = delete;

	void store(const int64_t val)//, std::memory_order _order = std::memory_order_seq_cst)
	{
		_seq.set(val);//,_order);
	}

	int64_t load()//std::memory_order _order = std::memory_order_seq_cst)
	{
		return _seq.get();// _order);
	}

	int64_t fetch_add(const int64_t increment)//, std::memory_order _order = std::memory_order_seq_cst)
	{
		return _seq.fetch_add(increment);// _order);
	}

	int64_t add_fetch(const int64_t increment)//, std::memory_order _order = std::memory_order_seq_cst)
	{
		return _seq.add(increment);// _order);
	}

private:
	//两边都补齐，以保证_seq不会与其它变量共享一个缓存行
	char _frontPadding[CACHELINE_PADDING_FOR_ATOMIC_INT64_SIZE];
	///std::atomic_int64_t _seq;
	atomic_int64 _seq;
	char _backPadding[CACHELINE_PADDING_FOR_ATOMIC_INT64_SIZE];
};

//对int64_t进行了封装，内存补齐保证_seq在一个缓存行中
class Sequence
{
public:
	Sequence(int64_t num = 0L) : _seq(num) {};
	~Sequence() {};
	Sequence(const Sequence&) = delete;
	Sequence(const Sequence&&) = delete;
	void operator=(const Sequence&) = delete;

	void store(const int64_t val)
	{
		_seq = val;
	}

	int64_t load()
	{
		return _seq;
	}

private:
	//两边都补齐，以保证_seq不会与其它变量共享一个缓存行
	char _frontPadding[CACHELINE_PADDING_FOR_INT64_SIZE];
	volatile int64_t _seq;
	char _backPadding[CACHELINE_PADDING_FOR_INT64_SIZE];
};

// //对int64_t进行了封装，内存补齐保证_seq在一个缓存行中
class ull_spsc_seq
{
public:
	ull_spsc_seq(int64_t num = 0L) : _seq(num) {};
	~ull_spsc_seq() {};
	ull_spsc_seq(const ull_spsc_seq &) = delete;
	ull_spsc_seq(const ull_spsc_seq &&) = delete;
	void operator=(const ull_spsc_seq &) = delete;

	void store(const int64_t val)
	{
		_seq = val;
	}

	int64_t load()
	{
		return _seq;
	}

private:
	//两边都补齐，以保证_seq不会与其它变量共享一个缓存行
	char _frontPadding[CACHELINE_PADDING_FOR_INT64_SIZE];
	volatile int64_t _seq;
	char _backPadding[CACHELINE_PADDING_FOR_INT64_SIZE];
};

//环形buffer默认大小,为了简化取余操作，这里的长度需要是2的n次方
const int64_t DefaultRingBufferSize = 1024;

//该disruptor类提供对环形buffer的操作
//写接口：WriteInBuf()
//读接口：(1)GetReadableSeq()获取可读的环形buffer槽位下标
//        (2)ReadFromBuf()获取可读的内容
//        (3)读完后调用FinishReading()
//注：读接口使用复杂，使用了BufConsumer类进行了封装（利用RAII）
template<class ValueType, int64_t N = DefaultRingBufferSize>
class Disruptor
{
public:
	///Disruptor() : _lastRead(-1L) , _lastWrote(-1L), _lastDispatch(-1L), _writableSeq(0L) , _stopWorking(0L){};
	Disruptor() : _lastRead(-1L), _lastWrote(-1L), _stopWorking(0L), _lastDispatch(-1L), _writableSeq(0L)
	{
		if (mlock(&_ringBuf, N))
		{
			perror("mlock");
			return;
		}
		size_t i;
		size_t page_size = getpagesize();
		for (i = 0; i < N; i += page_size)
		{
			_ringBuf[i] = {};
		}
	};
	~Disruptor() {};

	Disruptor(const Disruptor&) = delete;
	Disruptor(const Disruptor&&) = delete;
	void operator=(const Disruptor&) = delete;

	static_assert(((N > 0) && ((N& (~N + 1)) == N)),
		"RingBuffer's size must be a positive power of 2");

	//获取可写的buffer下标
	const int64_t GetWriteableSeq()
	{
		const int64_t writableSeq = _writableSeq.fetch_add(1);
		while (writableSeq - _lastRead.load() > N)
		{
			//等待策略
			if (_stopWorking.load() && empty())
			{
				return -1L;
			};
			// ull_pause();
		}

		return writableSeq;
	}

	//获取可写的buffer
	ValueType& GetWritableBuf(const int64_t writeableSeq)
	{
		if (writeableSeq < 0)
		{
			throw("error : incorrect seq for ring Buffer when GetWritableBuf(seq)!");
		}
		return _ringBuf[writeableSeq & (N - 1)];
	}

	//写完指定下标位置的buffer内容，交还下标位置使用权
	void FinishWriting(const int64_t seq)
	{
		if (seq < 0)
		{
			return;
		}

		while (seq - 1 != _lastWrote.load())
		{
			//等待策略,咱不能死等，让渡一下CPU
			// ull_pause();
		}

		_lastWrote.store(seq);
	};

	//向buffer中写内容
	void WriteInBuf(ValueType&& val)
	{
		const int64_t writableSeq = _writableSeq.fetch_add(1);
		while (writableSeq - _lastRead.load() > N)
		{ //等待策略
			if (_stopWorking.load())
				; //			throw std::runtime_error("writting when stopped disruptor");
				  // std::this_thread::yield();
				  //等待策略,咱不能死等，让渡一下CPU
			// ull_pause();
		}
		//写操作
		_ringBuf[writableSeq & (N - 1)] = val;

		while (writableSeq - 1 != _lastWrote.load())
		{
			//等待策略,咱不能死等，让渡一下CPU
			// ull_pause();
		}
		_lastWrote.store(writableSeq);
	};

	//向buffer中写内容
	void WriteInBuf(ValueType& val)
	{
		const int64_t writableSeq = _writableSeq.fetch_add(1);
		while (writableSeq - _lastRead.load() > N)
		{ //等待策略
			if (_stopWorking.load())
				; //			throw std::runtime_error("writting when stopped disruptor");
				  // std::this_thread::yield();
				  //等待策略,咱不能死等，让渡一下CPU
			// ull_pause();
		}
		//写操作
		_ringBuf[writableSeq & (N - 1)] = val;

		while (writableSeq - 1 != _lastWrote.load())
		{
			//等待策略,咱不能死等，让渡一下CPU
			// ull_pause();
		}
		_lastWrote.store(writableSeq);
	};

	//获取可读的buffer下标
	const int64_t GetReadableSeq()
	{
		const int64_t readableSeq = _lastDispatch.add_fetch(1);
		while (readableSeq > _lastWrote.load())
		{ //等待策略
			if (_stopWorking.load() && empty())
			{
				return -1L;
			}
			//等待策略, 咱不能死等，让渡一下CPU
			// ull_pause();
		}
		return readableSeq;
	};

	//读取指定下标位置的buffer内容
	ValueType& ReadFromBuf(const int64_t readableSeq)
	{
		if (readableSeq < 0)
		{
			throw("error : incorrect seq for ring Buffer when ReadFromBuf(seq)!");
		}
		return _ringBuf[readableSeq & (N - 1)];
	}

	//读取完指定下标位置的buffer内容，交还下标位置使用权
	void FinishReading(const int64_t seq)
	{
		if (seq < 0)
		{
			return;
		}

		while (seq - 1 != _lastRead.load())
		{
			//等待策略,咱不能死等，让渡一下CPU
			// ull_pause();

		}
		//_lastRead = seq;
		_lastRead.store(seq);
	};

	bool empty()
	{
		return _writableSeq.load() - _lastRead.load() == 1;
	}

	//通知disruptor停止工作，调用该函数后，若buffer已经全部处理完，那么获取可读下标时只会获取到-1L
	void stop()
	{
		//_stopWorking = true;
		_stopWorking.store(1L);
	}

	bool CanStop()
	{
		if (_stopWorking.load() && empty())
		{
			return true;
		}

		return false;
	}

private:
	//最后一个已读内容位置
	ull_spsc_seq _lastRead;

	//最后一个已写内容位置
	ull_spsc_seq _lastWrote;

	//disruptor是否停止工作
	ull_spsc_seq _stopWorking;

	//最后一个派发给消费者使用的槽位序号
	AtomicSequence _lastDispatch;

	//当前可写的槽位序号
	AtomicSequence _writableSeq;

	//环形buffer，为加快取余操作，N需要时2的n次幂
	std::array<ValueType, N> _ringBuf = {};
};

